package org.apache.flink.kubernetes.kubeclient.decorators;

import org.apache.flink.configuration.ConsumerManagerOptions;
import org.apache.flink.kubernetes.kubeclient.parameters.KubernetesMasterComponentParameters;
import org.apache.flink.kubernetes.kubeclient.services.HeadlessClusterIPService;

import io.fabric8.kubernetes.api.model.HasMetadata;
import io.fabric8.kubernetes.api.model.Service;

import java.io.IOException;
import java.util.Collections;
import java.util.List;

import static org.apache.flink.util.Preconditions.checkNotNull;

/**
 * Creates an internal Service which forwards the requests from the TaskManager(s) to the active
 * JobManager. Note that only the non-HA scenario relies on this Service for internal communication,
 * since in the HA mode, the TaskManager(s) directly connects to the JobManager via IP address.
 */
public class InternalServiceDecorator extends AbstractKubernetesStepDecorator {

    private final KubernetesMasterComponentParameters kubernetesJobManagerParameters;

    public InternalServiceDecorator(KubernetesMasterComponentParameters kubernetesJobManagerParameters) {
        this.kubernetesJobManagerParameters = checkNotNull(kubernetesJobManagerParameters);
    }

    @Override
    public List<HasMetadata> buildAccompanyingKubernetesResources() throws IOException {
        if (!kubernetesJobManagerParameters.isInternalServiceEnabled()) {
            return Collections.emptyList();
        }

        final String serviceName =
                getInternalServiceName(kubernetesJobManagerParameters.getClusterId());

        final Service headlessService =
                HeadlessClusterIPService.INSTANCE.buildUpInternalService(
                        kubernetesJobManagerParameters);

        // Set job manager address to namespaced service name
        final String namespace = kubernetesJobManagerParameters.getNamespace();
        kubernetesJobManagerParameters
                .getConfiguration()
                .setString(
                        ConsumerManagerOptions.ADDRESS,
                        getNamespacedInternalServiceName(serviceName, namespace));

        return Collections.singletonList(headlessService);
    }

    /** Generate name of the internal Service. */
    public static String getInternalServiceName(String clusterId) {
        return clusterId;
    }

    /** Generate namespaced name of the internal Service. */
    public static String getNamespacedInternalServiceName(String clusterId, String namespace) {
        return getInternalServiceName(clusterId) + "." + namespace;
    }
}
